/*
 * Copyright (C) 2018 The Asiainfo-Ability Authors
 *
 *      Licensed under the Apache License, Version 2.0 (the "License");
 *      you may not use this file except in compliance with the License.
 *      You may obtain a copy of the License at
 *
 *          http://www.apache.org/licenses/LICENSE-2.0
 *
 *      Unless required by applicable law or agreed to in writing, software
 *      distributed under the License is distributed on an "AS IS" BASIS,
 *      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *      See the License for the specific language governing permissions and
 *      limitations under the License.
 */

package org.asiainfo.ability.gateway.configuration;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.asiainfo.ability.gateway.cache.CacheService;
import org.asiainfo.ability.base.queue.OperHandler;
import org.asiainfo.ability.base.queue.QueueObserver;
import org.asiainfo.ability.base.queue.support.CacheEvent;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.cache.annotation.CachingConfigurerSupport;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.cache.interceptor.KeyGenerator;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.*;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import javax.annotation.PostConstruct;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;

/**
 * @Author: visen
 * @Date: 2018/2/2
 * @Description: 1 2 level cache, 1: local cache( fast Caffeine); 2: redis cache
 */
@Configuration
@EnableCaching
public class CacheConfig extends CachingConfigurerSupport {
    private static final Logger logger = LogManager.getLogger(CacheConfig.class);


    @PostConstruct
    public void init() {
        logger.error("======111");
    }
    @Bean
    public KeyGenerator keyGenerator() {
        return new KeyGenerator() {
            @Override
            public Object generate(Object target, Method method, Object... params) {
                StringBuilder sb = new StringBuilder();
                for (Object obj : params) {
                    sb.append(obj.toString());
                }
                return sb.toString();
            }
        };

    }





    @Bean
    CacheService cacheService(RedisConnectionFactory connectionFactory) {
        return new CacheService(connectionFactory);
    }


    @Bean
    @ConditionalOnMissingBean
    public CustomRedisCacheManager cacheManager(CacheService cacheService, QueueObserver queueObserver)
    {
        CustomRedisCacheManager cacheManager = new CustomRedisCacheManager(cacheService.getCacheClient());
        cacheManager.setDefaultExpiration(7 * 86400);
        cacheManager.setCachePrefix(new RedisCachePrefix() {
            private final RedisSerializer<String> serializer = new StringRedisSerializer();
            private final String delimiter = ":";

            @Override
            public byte[] prefix(String cacheName) {
                return this.serializer
                        .serialize(cacheName.concat(this.delimiter));
            }
        });
        queueObserver.subscribeEvent(CacheEvent.class, new OperHandler<CacheEvent>() {
            @Override
            public void process(CacheEvent event) {
                cacheManager.receiver(event.getName(), event.getKey());
            }
        });
        cacheManager.setUsePrefix(true);
        return cacheManager;
    }


    class CustomRedisCacheManager extends RedisCacheManager {


        public CustomRedisCacheManager(RedisOperations redisOperations) {
            super(redisOperations);

        }


        @SuppressWarnings("unchecked")
        @Override
        protected RedisCache createCache(String cacheName) {
            long expiration = computeExpiration(cacheName);
            return new CustomRedisCache(this, cacheName, (this.isUsePrefix() ? this.getCachePrefix().prefix(cacheName) : null), this.getRedisOperations(), expiration);
        }


        public void receiver(String name, String key) {
            CustomRedisCache cache = (CustomRedisCache) this.getCache(name);
            if (cache != null) {
                cache.cacheUpdate(key);
            }
        }

        public void publishMessage(String cacheName) {
            //this.getRedisOperations().convertAndSend(topicName, cacheName);
        }

    }

    class CustomRedisCache extends RedisCache {
        Cache<Object, ValueWrapper> local;
        CustomRedisCacheManager cacheManager;
        private String name;

        public CustomRedisCache(CustomRedisCacheManager cacheManager, String name, byte[] prefix,
                                RedisOperations<? extends Object, ? extends Object> redisOperations, long expiration) {
            super(name, prefix, redisOperations, expiration);
            this.name = name;
            this.cacheManager = cacheManager;
            local = Caffeine.newBuilder()
                    .expireAfterAccess(30, TimeUnit.MINUTES)
                    .maximumSize(512)
                    .build();
        }

        @Override
        public ValueWrapper get(Object key) {
            ValueWrapper wrapper = local.getIfPresent(key);
            if (wrapper != null) {
                return wrapper;
            } else {
                wrapper = super.get(key);
                if (wrapper != null) {
                    local.put(key, wrapper);
                }
                return wrapper;
            }

        }

        //redis-cli -h 127.0.0.1 -p 6903
        @Override
        public void put(final Object key, final Object value) {
            super.put(key, value);
            local.put(key, new RedisCacheElement(new RedisCacheKey(key), value));
        }

        @Override
        public void evict(Object key) {
            super.evict(key);
        }

        @Override
        public ValueWrapper putIfAbsent(Object key, final Object value) {
            ValueWrapper wrapper = super.putIfAbsent(key, value);
            return wrapper;
        }

        public void cacheUpdate(String fieldKey) {
            if (fieldKey != null) {
                local.invalidate(fieldKey);
            } else {
                local.invalidateAll();
            }
        }

    }


}

